package com.swhy

import com.utils._
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.codehaus.jackson.map.ObjectMapper

import java.net.URI
import java.util.UUID


object ChgeProdSaleRptBlxBly {
  def main(args: Array[String]): Unit = {
    if (args.length != 3) {
      println(
        """
          |com.swhy.CompanyInfo
          |params:
          | InputDate    刷数日期
          | OutputPath   输出路径
          | SplitNum     文件生成数量
        """.stripMargin)
      sys.exit()
    }
    val conf = new Configuration()
    val Array(inputDt, outputPath, splitNum) = args
    val sparkConf = new SparkConf().setAppName(this.getClass.getName)
    val sc = new SparkContext(sparkConf)
    val hiveContext = new HiveContext(sc)
    //改下方的sql为需要装换成jason的sql
    val allcompany =
      s"""
     select
      '132021/132026' as prod_code,
      '宝利鑫/宝利源' as prod_name ,
      a.asset_acct as capi_acc,
      b.cust_no,
      b.cust_name,
      nvl(b.mgr_serv,'') as mgr_cd,
      nvl(b.mgr_dev,'') as mgr_dev,
      nvl(b.mgr_sign,'') as mgr_sign,
      nvl(b.brch_org_level_cd_serv,'')  as brch_org_level_cd,
      nvl(b.brch_org_level_cd_dev,'')  as brch_org_level_cd_dev,
      nvl(b.brch_org_level_cd_sign,'')  as brch_org_level_cd_sign,
      nvl(b.brch_org_cd_serv,'')  as brch_org_cd,
      nvl(b.brch_org_cd_dev,'')  as brch_org_cd_dev,
      nvl(b.brch_org_cd_sign,'')  as brch_org_cd_sign,
      b.latest_risk_grad,
      b.cust_grad,
      b.cust_type,
      nvl(round(b.total_asset),0) as prod2_total_asset,
      round(nvl(f.xjlzc,0)) as  prod2_cap_bal,
      nvl(f.xjlzc_mx,'[]') as prod2_cap_bal_detail,
      nvl(a.hb_asset,0) as prod2_hb_asset,
      nvl(a.bjhg_asset,0) as prod2_bjhg_asset,
      nvl(a.dlwt_asset_1d,0) as prod2_dlwt_asset_1d,
      nvl(a.liulan_bjhy_flag_3m,'') as prod2_liulan_bjhy_flag_3m,
      nvl(a.liulan_nhg_flag_3m,'') as prod2_liulan_nhg_flag_3m,
      nvl(a.FUND_ASSET_RATE,0) as prod2_fund_asset_rate,
      nvl(a.pos_hb_flag,'') as prod2_pos_hb_flag,
      nvl(a.buy_blx_flag_3m,'') as prod2_buy_blx_flag_3m,
      nvl(a.buy_gznhg_flag_6m,'') as prod2_buy_gznhg_flag_6m,
      nvl(a.bjz_flag,'')  as prod2_bjz_flag,
      nvl(a.rg_count_3m,0) as prod2_rg_count_3m,
      nvl(a.repo_sh_Flag,'') as prod2_repo_sz_flag,
      nvl(a.Blxpl_Flag,'') as prod2_blxpl_flag,
      nvl(a.if_buy_bjhg,'') as prod2_if_buy_bjhg,
      nvl(a.if_no_blx_flag_asset_10000_3m,'') as prod2_if_no_blx_flag_asset_10000_3m,
      if(c.asset_acct is null ,'0','1') as prod2_buy_gp,
      if(d.asset_acct is null ,'0','1') as prod2_is_buy
      from app_data.app_rpt_blx_latent_acct_d_q a
      left join  app_data.app_dao_acct_info_d_q  b
      on  a.asset_acct =  b.capi_acc
      left join  (select distinct asset_acct from  sum_data.sum_t98_on_board_match_dtl_z a join pdm_data.pdm_t02_prod_q b on a.prod_id =b.prod_id where trim(prod_type_cd)  in('S00','S10','S30','S20','S22','S01','S11','S31','S21','S36','S21','S32','S33','S34','S35') and a.biz_date>=cast(regexp_replace(add_months(from_unixtime(unix_timestamp('${inputDt}','yyyyMMdd' ),'yyyy-MM-dd'),-12),'-','') as int) and b.biz_date=${inputDt})c
      on a.asset_acct=c.asset_acct
      left join app_data.app_daop_recmd_aft_purc_z d
      on a.asset_acct=d.asset_acct
      and d.prod_id='宝丽鑫/宝丽源'
      left join app_data.xjlzc_daop_temp  f
      on a.asset_acct=f.asset_acct
      where a.biz_date=${inputDt}  and
            b.biz_date=${inputDt}
      """.stripMargin
    print(allcompany)
    val frame  = hiveContext.sql(allcompany)
    val result = frame.rdd.map(x => {
      var cominfo = new StringBuilder()

      val prod_code = x.getAs[String]("prod_code")
      val prod_name = x.getAs[String]("prod_name")
      val cpai_acc = x.getAs[String]("capi_acc")
      val cust_no = x.getAs[String]("cust_no")
      val cust_name = x.getAs[String]("cust_name")
      val latest_risk_grad = x.getAs[String]("latest_risk_grad")
      val cust_grad = x.getAs[String]("cust_grad")
      val cust_type = x.getAs[String]("cust_type")

      val mgr_cd = x.getAs[String]("mgr_cd")
      val mgr_dev = x.getAs[String]("mgr_dev")
      val mgr_sign = x.getAs[String]("mgr_sign")
      val brch_org_cd = x.getAs[String]("brch_org_cd")
      val brch_org_cd_dev = x.getAs[String]("brch_org_cd_dev")
      val brch_org_cd_sign = x.getAs[String]("brch_org_cd_sign")
      val brch_org_level_cd = x.getAs[String]("brch_org_level_cd")
      val brch_org_level_cd_dev = x.getAs[String]("brch_org_level_cd_dev")
      val brch_org_level_cd_sign = x.getAs[String]("brch_org_level_cd_sign")

      val prod2_total_asset =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_total_asset"))
      val prod2_total_asset1=prod2_total_asset.bigDecimal
      val prod2_cap_bal_detail=x.getAs[String]("prod2_cap_bal_detail")
      val prod2_cap_bal =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_cap_bal"))
      val prod2_cap_bal1=prod2_cap_bal.bigDecimal
      val prod2_bjhg_asset =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_bjhg_asset"))
      val prod2_bjhg_asset1=prod2_bjhg_asset.bigDecimal
      val prod2_hb_asset =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_hb_asset"))
      val prod2_hb_asset1=prod2_hb_asset.bigDecimal
      val prod2_dlwt_asset_1d =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_dlwt_asset_1d"))
      val prod2_dlwt_asset_1d1=prod2_dlwt_asset_1d.bigDecimal
      val prod2_buy_gp =x.getAs[String]("prod2_buy_gp")
      val prod2_fund_asset_rate = scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_fund_asset_rate"))
      val prod2_fund_asset_rate1=prod2_fund_asset_rate.bigDecimal

      val prod2_liulan_bjhy_flag_3m = x.getAs[String]("prod2_liulan_bjhy_flag_3m")
      val prod2_liulan_nhg_flag_3m = x.getAs[String]("prod2_liulan_nhg_flag_3m")
      val prod2_pos_hb_flag = x.getAs[String]("prod2_pos_hb_flag")
      val prod2_is_buy = x.getAs[String]("prod2_is_buy")
      val prod2_buy_blx_flag_3m = x.getAs[String]("prod2_buy_blx_flag_3m")
      val prod2_buy_gznhg_flag_6m = x.getAs[String]("prod2_buy_gznhg_flag_6m")
      val prod2_bjz_flag = x.getAs[String]("prod2_bjz_flag")
      val prod2_rg_count_3m =scala.math.BigDecimal(x.getAs[java.math.BigDecimal]("prod2_rg_count_3m"))
      val prod2_rg_count_3m1=prod2_rg_count_3m.bigDecimal
      val prod2_repo_sz_flag = x.getAs[String]("prod2_repo_sz_flag")
      val prod2_blxpl_flag = x.getAs[String]("prod2_blxpl_flag")
      val prod2_if_buy_bjhg = x.getAs[String]("prod2_if_buy_bjhg")
      val prod2_if_no_blx_flag_asset_10000_3m = x.getAs[String]("prod2_if_no_blx_flag_asset_10000_3m")





      val _id = new ID()
      val index = new Index()
      val companyInfo = new ChgeProdSaleRptBlxBly()

      _id.set_id(UUID.randomUUID().toString)
      index.setIndex(_id)
      companyInfo.setProdCode(prod_code)
      companyInfo.setProdName(prod_name)
      companyInfo.setCapiAcc(cpai_acc)
      companyInfo.setCustNo(cust_no)
      companyInfo.setCustName(cust_name)
      companyInfo.setMgrCd(mgr_cd)
      companyInfo.setBrchOrgCd(brch_org_cd)
      companyInfo.setBrchOrgLevelCd(brch_org_level_cd)

      companyInfo.setMgrSign(mgr_sign)
      companyInfo.setMgrDev(mgr_dev)
      companyInfo.setBrchOrgCdDev(brch_org_cd_dev)
      companyInfo.setBrchOrgCdSign(brch_org_cd_sign)
      companyInfo.setBrchOrgLevelCdDev(brch_org_level_cd_dev)
      companyInfo.setBrchOrgLevelCdSign(brch_org_level_cd_sign)
      companyInfo.setProd2BuyGp(prod2_buy_gp)
      companyInfo.setProd2CapBalDetaail(prod2_cap_bal_detail)
      companyInfo.setLatestRiskGrad(latest_risk_grad)
      companyInfo.setCustGrad(cust_grad)
      companyInfo.setCustType(cust_type)
      companyInfo.setMgrDev(mgr_dev)
      companyInfo.setMgrSign(mgr_sign)
      companyInfo.setProd2TotalAsset(prod2_total_asset1)
      companyInfo.setProd2CapBal(prod2_cap_bal1)
      companyInfo.setProd2IsBuy(prod2_is_buy)
      companyInfo.setProd2DlwtAsset1d(prod2_dlwt_asset_1d1)
      companyInfo.setProd2FundAssetRate(prod2_fund_asset_rate1)
      companyInfo.setProd2LiulanBjhyFlag3m(prod2_liulan_bjhy_flag_3m)
      companyInfo.setProd2LiulanNhgFlag3m(prod2_liulan_nhg_flag_3m)
      companyInfo.setProd2PosHbFlag(prod2_pos_hb_flag)
      companyInfo.setProd2BuyBlxFlag3m(prod2_buy_blx_flag_3m)
      companyInfo.setProd2BuyGznhgFlag6m(prod2_buy_gznhg_flag_6m)
      companyInfo.setProd2BjzFlag(prod2_bjz_flag)
      companyInfo.setProd2RgCount3m(prod2_rg_count_3m1)
      companyInfo.setProd2RepoSzFlag(prod2_repo_sz_flag)
      companyInfo.setProd2BlxplFlag(prod2_blxpl_flag)
      companyInfo.setProd2IfBuyBjhg(prod2_if_buy_bjhg)
      companyInfo.setProd2IfNoBlxFlagAsset100003m(prod2_if_no_blx_flag_asset_10000_3m)
      companyInfo.setProd2BjhgAsset(prod2_bjhg_asset1)
      companyInfo.setProd2HbAsset(prod2_hb_asset1)
      val mapper = new ObjectMapper()

      val indexJson = mapper.writeValueAsString(index) //企业自增编码json串
      val companyInfoJson = mapper.writeValueAsString(companyInfo) //企业信息串

      cominfo = cominfo.append(indexJson + "\n" + companyInfoJson) //json结果串
      cominfo.toString()
    })
    val path = new Path(outputPath) //创建文本输出路径
    val fs = FileSystem.get(URI.create(outputPath), conf)
    if (fs.exists(path)) {
      fs.delete(path, true)
    }
    result.coalesce(splitNum.toInt, true).saveAsTextFile(outputPath)
    sc.stop()
  }
}
